package com.shujia.streeam

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo5transform {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("Demo1WordCount")
    conf.set("spark.sql.shuffle.partitions", "2")
    val sc = new SparkContext(conf)


    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._

    /**
      *
      * 创建 streamming上下文对象  需要指定间隔时间   延迟时间
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
      * 读取socket创建DS
      * nc -lk 8888
      *
      */

    val studentDS = ssc.socketTextStream("node1", 8888)


    /**
      * transform  将ds转换成rdd   需要返回一个rdd
      *
      */

    val resultDS = studentDS.transform(rdd => {


      val resultRDD = rdd.map(line => {
        val clazz = line.split(",")(4)
        (clazz, 1)
      })
        .reduceByKey(_ + _)


      //返回rdd
      resultRDD
    })


    resultDS.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
